/*
 *  Arrow
 *
 *  Created by David Giovannini on 1/5/07.
 *  Copyright 2007 Software by Jove. All rights reserved.
 *
 */
 
#include "SBJ/Arrow/Queue.h"
#include "SBJ/Arrow/MsgProxy.h"
#include "SBJ/Arrow/EnqueuePolicy.h"
#include "SBJ/Arrow/DequeuePolicy.h"
#include "SBJ/Thread/PlatformDetails.h"

#include <boost/bind.hpp>

#include <stdexcept>

namespace SBJ
{

namespace Arrow
{

Queue::Queue( size_t priorityCount, const IExceptionObserver::Ptr& excObserver,
				 EnqueuePolicy* enqueuePolicy, DequeuePolicy* dequeuePolicy )
: m_started( false )
, m_queues( priorityCount )
, m_maxSizes( priorityCount )
, m_excObserver( excObserver )
, m_enqueuePolicy( enqueuePolicy ? enqueuePolicy : new EnqueuePolicy )
, m_dequeuePolicy( dequeuePolicy ? dequeuePolicy : new DequeuePolicy )
, m_hasInterest( false )
{
	if ( ! m_dequeuePolicy->getRequiredPriority().isInRange( priorityCount - 1 ) )
	{
		throw std::out_of_range( "Queue priorities not in range of Dequeue Policy" );
	}
}

Queue::~Queue( void )
{
	stop();
	flush();
}

bool
Queue::start( void )
{
    if ( m_started == false )
	{
		m_enqueueStat.start();

		m_started = true;
		m_dequeueThread.reset(
			new boost::thread(
				boost::bind( &Queue::threadWork, boost::ref( *this ) ) ) );
	}
	return ( m_started == true );
}

bool
Queue::stop( bool join )
{
	if ( m_started == true )
	{
		m_enqueueStat.stop();

		m_started = false;
		m_queueCondition.notify_one();
		if ( join == true )
		{
			m_dequeueThread->join();
			m_dequeueThread.reset();
		}
	}
	return ( m_started ==  false );
}

void
Queue::flush( void )
{
	boost::recursive_mutex::scoped_lock lock( m_queueGuard );
	
	KeyedNodes::iterator iter( m_knownNodes.begin() );
	while ( iter != m_knownNodes.end() )
	{
		if ( ! iter->second->isQueued() )
		{
			delete iter->second;
		}
		++iter;
	}
	m_knownNodes.clear();

	for ( size_t i = 0; i < m_queues.size(); ++i )
	{
		m_queues[ i ].clear();
	}
}

void
Queue::onMessage( Message& msg )
{
	// Is this a valid message
	if ( validityCheck( msg ) == false ) return;

	// Determine the correct priority
	size_t priority = determinePriority( msg );
	if ( priority == -1 ) return;

	// Box the message
	Message* box = boxMessage( msg );

	// Enqueue it
	if ( box != 0 )
	{
		enqueueNode( box, priority );
	}
}

bool
Queue::validityCheck( Message& msg )
{
	const MsgProxy& proxy = msg.getTopic().getMsgProxy();
	if ( proxy.isValid( msg ) == false )
	{
		try
		{
			msg.dropped( Message::dropped_InvalidTopic );
		}
		catch ( std::exception& exception )
		{
			if ( m_excObserver ) m_excObserver->onException( exception );
		}
		catch ( ... )
		{
			if ( m_excObserver ) m_excObserver->onException( std::logic_error( "Unknown error on Message dropped" ) );
		}
		m_enqueueStat.accumeInvalid();
		return false;
	}
	return true;
}

size_t
Queue::determinePriority( Message& msg )
{
	const MsgProxy& proxy = msg.getTopic().getMsgProxy();
	const size_t lowestPriority = ( m_queues.size() - 1 );

	size_t priority = msg.getPriority();
	if ( priority > lowestPriority )
	{
		if ( m_enqueuePolicy->dropOutOfRange() == true )
		{
			proxy.dropped( msg, Message::dropped_PriorityTooLow );
			m_enqueueStat.accumeDrop();
			return -1;
		}
		else
		{
			priority = lowestPriority;
		}
	}
	return priority;
}

Message*
Queue::boxMessage( Message& msg )
{
	// Find the collisions
	Message* local = 0;

	Message& newMsg = msg;
	const Topic& newTopic = newMsg.getTopic();
	const MsgProxy& newProxy = newTopic.getMsgProxy();

	// Do we not care about consumption or reuse?
	if ( newProxy.fullyIndependent() == true )
	{
		local = newProxy.duplicate( newMsg ); // 0 on exception
		msg.setIndependent();
		m_enqueueStat.accumeIndependent();
	}
	// Find box for collisions
	else
	{
		// Look it up
		boost::recursive_mutex::scoped_lock lock( m_collisionGuard );
		std::pair< KeyedNodes::iterator, bool > result = m_knownNodes.insert(
			std::make_pair( newTopic, (Message*)0 ) );
		Message*& box = result.first->second;

		// If discovered
		if ( box == 0 )
		{
			box = newProxy.duplicate( newMsg ); // 0 on exception
			m_enqueueStat.accumeDiscovery();
			local = box;
			return local; // Bypass drop check
		}

		Message& oldMsg = *box;
		const Topic& oldTopic = oldMsg.getTopic();
		const MsgProxy& oldProxy = oldMsg.getTopic().getMsgProxy();

		// If the new message is a different type from the old
		if ( oldTopic.isSameMsgType( newTopic ) == false )
		{
			box->setIndependent();
			box = newProxy.duplicate( newMsg ); // 0 on exception
			m_enqueueStat.accumeVariant();
			local = box;
		}
		// If a collision of equal types
		else if ( box->isQueued() == true )
		{
			boost::tribool consumes = oldProxy.doesConsume( oldMsg, newMsg );
			// if consumes
			if ( consumes == true )
			{
				oldProxy.consume( oldMsg, newMsg );
				m_enqueueStat.accumeConsume();
				return 0;
			}
			// if collision
			else if ( !consumes )
			{
				box->setIndependent();
				box = newProxy.duplicate( newMsg ); // 0 on exception
				m_enqueueStat.accumeCollision();
				local = box;
			}
			else
			{
				local = 0; // error on doesConsume
			}
		}
		// If non-collision of equal types
		else
		{
			box = oldProxy.replace( box, newMsg );
			m_enqueueStat.accumeReuse();
			local = box;
		}
	}
	// Drop if no box created or found
	if ( local == 0 )
	{
		msg.getTopic().getMsgProxy().dropped( msg, Message::dropped_UncaughtException );
		m_enqueueStat.accumeDrop();
	}
	return local;
}

void
Queue::enqueueNode( Message* box, size_t priority )
{	
	Message& msg = *box;
	MsgQueue* queue = &m_queues[ priority ];
	
	boost::recursive_mutex::scoped_lock lock( m_queueGuard );
	// Quick check to see if we hit a limit
	if ( false && queue->size() + 1 >= m_maxSizes[ priority ] )
	{
		// Find the next available queue
		while ( priority < m_queues.size() && queue->size() + 1 >= m_maxSizes[ priority ] && m_enqueuePolicy->moveToNextOnFull( priority ) )
		{
			++priority;
			queue = &m_queues[ priority ];
		}
		// If we passed all queues
		if ( priority >= m_queues.size() )
		{
			msg.getTopic().getMsgProxy().dropped( msg, Message::dropped_FellOffQueue );
			m_enqueueStat.accumeDrop();
			return;
		}
		// If we have a queue determine the policy
		switch ( m_enqueuePolicy->replaceOnFull( priority ) )
		{
			// This queue just drops messages
			case EnqueuePolicy::eSide_none:
			{
				msg.getTopic().getMsgProxy().dropped( msg, Message::dropped_FellOffQueue );
				m_enqueueStat.accumeDrop();
				return;
			}
			// This queue drops the front
			case EnqueuePolicy::eSide_front:
			{
				if ( queue->empty() == false )
				{
					MsgQueue adopter;
					adopter.push_back( queue->front() );
					adopter.front()->getTopic().getMsgProxy().dropped( msg, Message::dropped_FellOffQueue );
					m_enqueueStat.accumeDrop();
				}
				break;
			}
			// This queue drops the back
			case EnqueuePolicy::eSide_back:
			{
				if ( queue->empty() == false )
				{
					MsgQueue adopter;
					adopter.push_back( queue->back() );
					adopter.front()->getTopic().getMsgProxy().dropped( msg, Message::dropped_FellOffQueue );
					m_enqueueStat.accumeDrop();
				}
				break;
			}
		}
	}
			
	// Determine if the dequeue thread needs to wake up
	bool wasEmpty = queue->empty();
	// We always just push back - all other behaviors are variable
	queue->push_back( box );
	// Wake up the other thread
	if ( wasEmpty && m_router && m_dequeuePolicy->getRequiredPriority().isInRange( priority ) )
	{
		m_wakeupPriority = priority;
		m_queueCondition.notify_one();
	}
}

bool
Queue::addRouter( const IRouter::Ptr& router )
{
	m_router = router;
	return true;
}

bool
Queue::removeRouter( const IRouter::Ptr& router )
{
	if ( m_router == router )
	{
		m_router.reset();
		return true;
	}
	return false;
}

void
Queue::threadWork( void )
{
	PlatformDetails::ThreadGuard threadGuard;
	m_dequeueStat.start();
	while ( m_started == true )
	{
		try
		{
			dequeue();
		}
		catch ( std::exception& exc )
		{
			if ( m_excObserver ) m_excObserver->onException( exc );
		}
		catch ( ... )
		{
			if ( m_excObserver ) m_excObserver->onException( std::logic_error( "unknown excption on dequeue" ) );
		}
	}
	m_dequeueStat.stop();
}

Message::Priority
Queue::haveMessage( Message::Priority high, Message::Priority low ) const
{
	for ( Message::Priority i = high; i < m_queues.size() && i <= low; ++i )
	{
		if ( m_queues[ i ].empty() == false )
		{
			return i;
		}
	}
	return Message::unknown_priority;
}

bool
Queue::dequeue( void )
{
	MsgQueue adopter;

	Message* node = 0;
	{
		boost::recursive_mutex::scoped_lock lock( m_queueGuard );
		DequeuePolicy::Range range = m_dequeuePolicy->getRequiredPriority();
		Message::Priority priority = haveMessage( range.high, range.low );
		// Handle the empty case 
		if ( priority == Message::unknown_priority )
		{
			m_dequeueStat.accumeSleep();
			m_queueCondition.wait( lock );

			if ( m_started == false )
			{
				return false;
			}
			else
			{
				priority = m_wakeupPriority;
			}
		}

		DequeuePolicy::Request request = m_dequeuePolicy->selectNode( priority, m_queues );

		// Transfer the node to the dequeue list
		if ( request.dequeueFrom != Message::unknown_priority )
		{
			if ( request.side == DequeuePolicy::eSide_front )
				node = m_queues[ request.dequeueFrom ].front();
			else  if ( request.side == DequeuePolicy::eSide_back )
				node = m_queues[ request.dequeueFrom ].back();
		}

		// Enqueue it locally so it becomes "locked"
		if ( node )
		{
			adopter.push_back( node );
		}
		else
		{
			return false;
		}
	}

	// Send it off
	if ( m_router )
	{
		m_dequeueStat.accumeRouted();
		try
		{
			m_router->onMessage( *node );
		}
		catch ( std::exception& exc )
		{
			if ( m_excObserver ) m_excObserver->onException( exc );
		}
		catch ( ... )
		{
			if ( m_excObserver ) m_excObserver->onException( std::logic_error( "unknown excption on route" ) );
		}
	}
	
	if ( node->isIndependent() == false )
	{
		adopter.pop_back();
	}
	// else have the adopter clear delete it

	return true;
}

}

}
